package cn.dmp.tags

import cn.dmp.util.Constans
import com.typesafe.config.ConfigFactory
import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

/**
  * 将每一行数据进行标签化处理
  * Created by Administrator on 2018/4/28.
  */
object Tags4Context {

  def main(args: Array[String]): Unit = {

    val config = ConfigFactory.load()

    val conf = new SparkConf()
      .setMaster("local[*]").setAppName(s"${this.getClass.getSimpleName}")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf) //spark离线job的入口

    val sQLContext = new SQLContext(sc)
    val dataFrame = sQLContext.read.parquet(config.getString("InputPath"))

    //读取appdic文件
    val appdict = sc.textFile(config.getString("appdict.path"))
    val appAndId = appdict.map(t =>t.split("\t",-1)).filter(_.length>=5)
      .map(arr=>(arr(4),arr(1))).collect().toMap
    //将字典文件广播出去
    val appDicBD = sc.broadcast(appAndId)

    //读取stop文件并广播
    val stopWords: Map[String, Int] = sc.textFile(config.getString("stopWords.path")).map(f => (f, 0))
      .collect().toMap
    val stopBD = sc.broadcast(stopWords)

    dataFrame.where(Constans.havaUserIdLog).map({row =>

      //广告标签
      val tagsAD = Tags4AD.makeTags(row)
      //APP标签
      val tagsAPP = Tags4App.makeTags(row,appDicBD)
      //设备
      val tagsDevice = Tags4Device.makeTags(row)
      //关键字
      val tagsKeyW = Tags4KeyW.makeTags(row,stopBD)
      //地域标签
      val tagsPro = Tags4ProCity.makeTags(row)

      (getUserId(row),(tagsAD++tagsAPP++tagsDevice++tagsKeyW++tagsPro).toList)
    }).reduceByKey({
      (list1,list2)=> (list1 ++ list2).groupBy(t=>t._1).mapValues(f=>f.foldLeft(0)(_ +_._2)).toList
        //.mapValues(f=>f.map(_._2).sum).toList
    }).foreach(println)

    //sc.stop()
  }

  def getUserId(row: Row): String = {
    row match {
      case row if !row.getAs[String]("imei").isEmpty => "IM:"+row.getAs[String]("imei")
      case row if !row.getAs[String]("mac").isEmpty => "MAC:"+row.getAs[String]("mac")
      case row if !row.getAs[String]("idfa").isEmpty => "IDF:"+row.getAs[String]("idfa")
      case row if !row.getAs[String]("openudid").isEmpty => "OPE:"+row.getAs[String]("openudid")
      case row if !row.getAs[String]("androidid").isEmpty => "AND:"+row.getAs[String]("androidid")

      case row if !row.getAs[String]("imeimd5").isEmpty => "IMm5:"+row.getAs[String]("imeimd5")
      case row if !row.getAs[String]("macmd5").isEmpty => "MACm5:"+row.getAs[String]("macmd5")
      case row if !row.getAs[String]("idfamd5").isEmpty => "IDFm5:"+row.getAs[String]("idfamd5")
      case row if !row.getAs[String]("openudidmd5").isEmpty => "OPEm5:"+row.getAs[String]("openudidmd5")
      case row if !row.getAs[String]("androididmd5").isEmpty => "ANDm5:"+row.getAs[String]("androididmd5")

      case row if !row.getAs[String]("imeisha1").isEmpty => "IMs1:"+row.getAs[String]("imeisha1")
      case row if !row.getAs[String]("macsha1").isEmpty => "MACs1:"+row.getAs[String]("macsha1")
      case row if !row.getAs[String]("idfasha1").isEmpty => "IDFs1:"+row.getAs[String]("idfasha1")
      case row if !row.getAs[String]("openudidsha1").isEmpty => "OPEs1:"+row.getAs[String]("openudidsha1")
      case row if !row.getAs[String]("androididsha1").isEmpty => "ANDs1:"+row.getAs[String]("androididsha1")
    }

  }
}